🎖️GitЯра🎖️
Commit e12c4a579d97dab1e8646fcd7bcbe710406d1076
Parents : 80bed50
Author : Jeremiah K <17190268+jeremiah-k@users.noreply.github.com>
Signature : Signature validation error
Date : 2026-07-07T14:58:43-05:00
Committer : GitHub <noreply@github.com>
Date : 2026-07-07T19:58:43Z
fix(runtime): harden BLE profile, OTA setup, and DB access (#6126)
Changes
7 files changed, 306 insertions(+), 65 deletions(-)
Diff
diff --git a/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/BleConnection.kt b/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/BleConnection.kt
index 35cb3bee32..c38f38e7d5 100644
--- a/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/BleConnection.kt
+++ b/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/BleConnection.kt
@@ -62,7 +62,11 @@ interface BleConnection {
/** Disconnects from the current device. */
suspend fun disconnect()
- /** Executes a block within a discovered profile. */
+ /**
+ * Executes [setup] after the requested BLE profile is available. This is shared by radio, OTA, and DFU flows, so
+ * implementations should keep profile-entry setup bounded and avoid transfer-hot-path work here when a caller can
+ * reuse an already discovered [BleService].
+ */
suspend fun <T> profile(
serviceUuid: Uuid,
timeout: Duration = 30.seconds,
diff --git a/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/KableBleConnection.kt b/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/KableBleConnection.kt
index 506c8be82b..e143eb2672 100644
--- a/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/KableBleConnection.kt
+++ b/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/KableBleConnection.kt
@@ -29,15 +29,20 @@ import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.NonCancellable
import kotlinx.coroutines.TimeoutCancellationException
+import kotlinx.coroutines.async
+import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.delay
+import kotlinx.coroutines.ensureActive
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.job
import kotlinx.coroutines.withContext
import kotlinx.coroutines.withTimeout
+import org.meshtastic.core.common.util.ioDispatcher
import kotlin.concurrent.Volatile
import kotlin.time.Duration
import kotlin.time.Duration.Companion.seconds
@@ -245,6 +250,7 @@ class KableBleConnection(private val scope: CoroutineScope, private val loggingC
_deviceFlow.emit(null)
}
+ @Suppress("ThrowsCount")
override suspend fun <T> profile(
serviceUuid: Uuid,
timeout: Duration,
@@ -253,7 +259,34 @@ class KableBleConnection(private val scope: CoroutineScope, private val loggingC
val p = peripheral ?: error("Not connected")
val cScope = connectionScope ?: error("No active connection scope")
val service = KableBleService(p, serviceUuid)
- return withTimeout(timeout) { cScope.setup(service) }
+ return withTimeout(timeout) {
+ // Shared BLE profile guard: wait for Kable service discovery before handing out the service, and map a
+ // connection-scope shutdown during caller setup to NotConnectedException instead of waiting for timeout.
+ withContext(ioDispatcher) {
+ val profileExecution = async {
+ p.services.first { it != null }
+ cScope.setup(service)
+ }
+
+ val disconnectHandle =
+ cScope.coroutineContext.job.invokeOnCompletion {
+ profileExecution.cancel(CancellationException("Connection lost during BLE profile execution"))
+ }
+
+ try {
+ profileExecution.await()
+ } catch (e: CancellationException) {
+ currentCoroutineContext().ensureActive()
+ if (!cScope.coroutineContext.job.isActive) {
+ throw NotConnectedException("Connection lost during BLE profile execution")
+ }
+ throw e
+ } finally {
+ disconnectHandle.dispose()
+ profileExecution.cancel()
+ }
+ }
+ }
}
override fun maximumWriteValueLength(writeType: BleWriteType): Int? = peripheral?.negotiatedMaxWriteLength()
diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt
index bbfe75cbee..f098ad4622 100644
--- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt
+++ b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt
@@ -31,7 +31,9 @@ import kotlinx.coroutines.Job
import kotlinx.coroutines.NonCancellable
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
+import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.delay
+import kotlinx.coroutines.ensureActive
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.StateFlow
@@ -279,36 +281,74 @@ open class DatabaseManager(
Logger.d { "Closed inactive database ${anonymizeDbName(dbName)} to free connections" }
}
- private val limitedIo = dispatchers.io.limitedParallelism(4)
+ // Short-term runtime containment: route withDb entry through a single-lane dispatcher to narrow the Room/SQLite
+ // connection-pool churn window seen during device/firmware update flows. Room suspend DAOs may continue on Room's
+ // own executor after suspension, so this is not a strict global DB-I/O serialization guarantee. Preserve bounded
+ // one-shot DB-critical blocks through cancellation, then re-check cancellation so stale callers do not continue
+ // after the DB releases. Long-lived Flow/Paging reads must stay out of withDb; revisit after direct currentDb.value
+ // callers are audited and safe DB concurrency can be restored.
+ private val limitedIo = dispatchers.io.limitedParallelism(1)
/** Execute [block] with the current DB instance. Retries once if the pool closes during a DB switch. */
@Suppress("TooGenericExceptionCaught")
- override suspend fun <T> withDb(block: suspend (MeshtasticDatabase) -> T): T? = withContext(limitedIo) {
- val db = _currentDb.value ?: return@withContext null
+ override suspend fun <T> withDb(block: suspend (MeshtasticDatabase) -> T): T? {
+ val queuedAt = nowMillis
+ return withContext(limitedIo) {
+ val queuedMillis = nowMillis - queuedAt
+ if (queuedMillis >= WITH_DB_SLOW_OPERATION_MS) {
+ Logger.w { "withDb waited ${queuedMillis}ms for the temporary DB containment lane" }
+ }
+
+ val startedAt = nowMillis
+ try {
+ withCurrentDb(block)
+ } finally {
+ val elapsedMillis = nowMillis - startedAt
+ if (elapsedMillis >= WITH_DB_SLOW_OPERATION_MS) {
+ Logger.w {
+ "withDb callback took ${elapsedMillis}ms on the temporary DB containment lane; persistent " +
+ "slow logs indicate DB access path should be revisited"
+ }
+ }
+ }
+ }
+ }
+
+ @Suppress("ReturnCount", "ThrowsCount", "TooGenericExceptionCaught")
+ private suspend fun <T> withCurrentDb(block: suspend (MeshtasticDatabase) -> T): T? {
+ val db = _currentDb.value ?: return null
val active = currentDbName
markLastUsed(active)
try {
- block(db)
+ return runCancellableDbBlock(db, block)
} catch (e: CancellationException) {
throw e // Preserve structured concurrency cancellation propagation.
} catch (e: Exception) {
// If the active database switched while we held a reference to the old one,
// and the exception indicates a closed pool/connection, retry with the new DB.
val retryDb = _currentDb.value
- if (retryDb != null && retryDb !== db && isDbClosedException(e)) {
- Logger.w { "withDb: database closed during switch (${e.message}), retrying with current DB" }
- try {
- block(retryDb)
- } catch (retryEx: Exception) {
- retryEx.addSuppressed(e)
- throw retryEx
- }
- } else {
- throw e
+ if (retryDb == null || retryDb === db || !isDbClosedException(e)) throw e
+
+ Logger.w { "withDb: database closed during switch (${e.message}), retrying with current DB" }
+ return try {
+ runCancellableDbBlock(retryDb, block)
+ } catch (retryCancel: CancellationException) {
+ throw retryCancel
+ } catch (retryEx: Exception) {
+ retryEx.addSuppressed(e)
+ throw retryEx
}
}
}
+ private suspend fun <T> runCancellableDbBlock(db: MeshtasticDatabase, block: suspend (MeshtasticDatabase) -> T): T {
+ // Keep withDb callbacks bounded and one-shot: NonCancellable can hold the containment lane until this returns.
+ currentCoroutineContext().ensureActive()
+ val result = withContext(NonCancellable) { block(db) }
+ currentCoroutineContext().ensureActive()
+ return result
+ }
+
private fun isDbClosedException(e: Exception): Boolean = generateSequence<Throwable>(e) { it.cause }
.any { throwable ->
val msg = throwable.message?.lowercase() ?: return@any false
@@ -317,6 +357,7 @@ open class DatabaseManager(
private companion object {
private const val BACKFILL_COLD_START_DELAY_MS = 2_000L
+ private const val WITH_DB_SLOW_OPERATION_MS = 1_000L
val DB_TERMS = listOf("pool", "database", "connection", "sqlite")
}
diff --git a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeBle.kt b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeBle.kt
index b792ee9b17..040a9e0751 100644
--- a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeBle.kt
+++ b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeBle.kt
@@ -120,6 +120,9 @@ class FakeBleConnection :
/** Number of times [connectAndAwait] has been invoked (including failures). */
var connectAndAwaitCalls: Int = 0
+ /** Number of times [profile] has been invoked. */
+ var profileCalls: Int = 0
+
/** Externally simulate a remote disconnect (e.g. node power-cycle) for tests that exercise reconnect. */
fun simulateRemoteDisconnect(reason: DisconnectReason = DisconnectReason.Timeout) {
_connectionState.value = BleConnectionState.Disconnected(reason)
@@ -168,6 +171,7 @@ class FakeBleConnection :
timeout: Duration,
setup: suspend CoroutineScope.(BleService) -> T,
): T {
+ profileCalls++
if (serviceUuid in missingServices) {
throw NoSuchElementException("Service $serviceUuid not found")
}
diff --git a/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransport.kt b/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransport.kt
index cd9d7b781c..c893f55256 100644
--- a/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransport.kt
+++ b/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransport.kt
@@ -20,18 +20,27 @@ import co.touchlab.kermit.Logger
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.CoroutineStart
+import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.TimeoutCancellationException
import kotlinx.coroutines.cancel
import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.delay
+import kotlinx.coroutines.ensureActive
import kotlinx.coroutines.flow.catch
+import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.withTimeoutOrNull
import org.meshtastic.core.ble.BleCharacteristic
import org.meshtastic.core.ble.BleConnectionFactory
import org.meshtastic.core.ble.BleConnectionState
import org.meshtastic.core.ble.BleDevice
import org.meshtastic.core.ble.BleScanner
+import org.meshtastic.core.ble.BleService
import org.meshtastic.core.ble.BleWriteType
import org.meshtastic.core.ble.MeshtasticBleConstants.OTA_NOTIFY_CHARACTERISTIC
import org.meshtastic.core.ble.MeshtasticBleConstants.OTA_SERVICE_UUID
@@ -56,89 +65,139 @@ class BleOtaTransport(
private val otaChar = BleCharacteristic(OTA_WRITE_CHARACTERISTIC)
private val txChar = BleCharacteristic(OTA_NOTIFY_CHARACTERISTIC)
- private val responseChannel = Channel<String>(Channel.UNLIMITED)
+ @Volatile private var otaService: BleService? = null
+
+ @Volatile private var responseChannel = Channel<String>(Channel.UNLIMITED)
// Written from the connectionState collector (Dispatchers.Default) and read by the streaming loop's
// connection-loss guard (Dispatchers.IO); @Volatile ensures the guard sees a mid-transfer disconnect.
@Volatile private var isConnected = false
+ @Volatile private var notificationJob: Job? = null
+
+ @Volatile private var connectionStateJob: Job? = null
+
/** Scan for the device by MAC address (or MAC+1 for OTA mode) with retries. */
private suspend fun scanForOtaDevice(): BleDevice? {
val otaAddress = calculateMacPlusOne(address)
val targetAddresses = setOf(address, otaAddress)
- Logger.i { "BLE OTA: Will match addresses: $targetAddresses" }
+ Logger.i { "BLE OTA: Will match target OTA device addresses" }
return scanForBleDevice(scanner = scanner, tag = "BLE OTA", serviceUuid = OTA_SERVICE_UUID) {
it.address in targetAddresses
}
}
- @Suppress("MagicNumber")
+ @Suppress("MagicNumber", "LongMethod")
override suspend fun connect(): Result<Unit> = safeCatching {
+ otaService = null
+ notificationJob?.cancel()
+ notificationJob = null
+ connectionStateJob?.cancel()
+ connectionStateJob = null
+ isConnected = false
+ responseChannel.close()
+ val connectResponseChannel = Channel<String>(Channel.UNLIMITED)
+ responseChannel = connectResponseChannel
+
Logger.i { "BLE OTA: Waiting $REBOOT_DELAY for device to reboot into OTA mode..." }
delay(REBOOT_DELAY)
- Logger.i { "BLE OTA: Connecting to $address using Kable..." }
+ Logger.i { "BLE OTA: Connecting to OTA device using Kable..." }
val device =
scanForOtaDevice()
?: throw OtaProtocolException.ConnectionFailed(
- "Device not found at address $address. " +
- "Ensure the device has rebooted into OTA mode and is advertising.",
+ "Device not found. Ensure the device has rebooted into OTA mode and is advertising.",
)
- bleConnection.connectionState
- .onEach { state ->
- Logger.d { "BLE OTA: Connection state changed to $state" }
- isConnected = state is BleConnectionState.Connected
- }
- .launchIn(transportScope)
+ connectionStateJob =
+ bleConnection.connectionState
+ .onEach { state ->
+ Logger.d { "BLE OTA: Connection state changed to $state" }
+ isConnected = state is BleConnectionState.Connected
+ }
+ .launchIn(transportScope)
try {
val finalState = bleConnection.connectAndAwait(device, CONNECTION_TIMEOUT)
if (finalState is BleConnectionState.Disconnected) {
- Logger.w { "BLE OTA: Failed to connect to ${device.address} (state=$finalState)" }
- throw OtaProtocolException.ConnectionFailed("Failed to connect to device at address ${device.address}")
+ Logger.w { "BLE OTA: Failed to connect to OTA device (state=$finalState)" }
+ throw OtaProtocolException.ConnectionFailed("Failed to connect to OTA device")
}
} catch (@Suppress("SwallowedException") e: kotlinx.coroutines.TimeoutCancellationException) {
- Logger.w { "BLE OTA: Timed out waiting to connect to ${device.address}. Error: ${e.message}" }
- throw OtaProtocolException.Timeout("Timed out connecting to device at address ${device.address}")
+ currentCoroutineContext().ensureActive()
+ Logger.w { "BLE OTA: Timed out waiting to connect to OTA device. Error: ${e.message}" }
+ throw OtaProtocolException.Timeout("Timed out connecting to OTA device")
}
- Logger.i { "BLE OTA: Connected to ${device.address}, discovering services..." }
+ Logger.i { "BLE OTA: Connected to OTA device, discovering services..." }
- bleConnection.profile(OTA_SERVICE_UUID) { service ->
- // Log negotiated MTU for diagnostics
- val maxLen = bleConnection.maximumWriteValueLength(BleWriteType.WITHOUT_RESPONSE)
- Logger.i { "BLE OTA: Service ready. Max write value length: $maxLen bytes" }
+ try {
+ bleConnection.profile(OTA_SERVICE_UUID) { service ->
+ service.requireOtaCharacteristics()
+
+ // Log negotiated MTU for diagnostics
+ val maxLen = bleConnection.maximumWriteValueLength(BleWriteType.WITHOUT_RESPONSE)
+ Logger.i { "BLE OTA: Service ready. Max write value length: $maxLen bytes" }
+
+ // Collect notification responses. Kable writes the CCCD descriptor as part of flow collection
+ // startup. Prefer the onSubscription readiness callback, but bound the wait because some OTA loaders
+ // never report the CCCD write completion even though notifications still flow.
+ val notificationsReady = CompletableDeferred<Unit>()
+ notificationJob =
+ launch(start = CoroutineStart.UNDISPATCHED) {
+ service
+ .observe(txChar) {
+ Logger.d { "BLE OTA: TX characteristic subscribed" }
+ notificationsReady.complete(Unit)
+ }
+ .onEach { notifyBytes ->
+ try {
+ val response = notifyBytes.decodeToString()
+ Logger.d { "BLE OTA: Received response: $response" }
+ connectResponseChannel.trySend(response)
+ } catch (@Suppress("TooGenericExceptionCaught") e: Exception) {
+ Logger.e(e) { "BLE OTA: Failed to decode response bytes" }
+ }
+ }
+ .catch { e ->
+ Logger.e(e) { "BLE OTA: Error in TX characteristic notification flow" }
+ if (!notificationsReady.isCompleted) {
+ notificationsReady.completeExceptionally(e)
+ }
+ connectResponseChannel.close(e)
+ }
+ .collect()
+ }
- // Collect responses. onSubscription fires when the CCCD write completes — a precise readiness
- // signal; the settle below is a conservative cushion.
- val subscribed = CompletableDeferred<Unit>()
- service
- .observe(txChar) {
- Logger.d { "BLE OTA: TX characteristic subscribed" }
- subscribed.complete(Unit)
- }
- .onEach { notifyBytes ->
- try {
- val response = notifyBytes.decodeToString()
- Logger.d { "BLE OTA: Received response: $response" }
- responseChannel.trySend(response)
- } catch (@Suppress("TooGenericExceptionCaught") e: Exception) {
- Logger.e(e) { "BLE OTA: Failed to decode response bytes" }
+ val confirmed = withTimeoutOrNull(SUBSCRIPTION_SETTLE) { notificationsReady.await() } != null
+ if (confirmed) {
+ Logger.i { "BLE OTA: TX notifications subscribed" }
+ } else {
+ Logger.w {
+ "BLE OTA: TX notification subscription not confirmed after $SUBSCRIPTION_SETTLE; " +
+ "continuing with bounded settle fallback"
}
}
- .catch { e ->
- if (!subscribed.isCompleted) subscribed.completeExceptionally(e)
- Logger.e(e) { "BLE OTA: Error in TX characteristic subscription" }
- }
- .launchIn(this)
-
- subscribed.await()
- // Conservative settle after CCCD confirmation before issuing commands.
- delay(SUBSCRIPTION_SETTLE)
- Logger.i { "BLE OTA: Service discovered and ready" }
+ otaService = service
+ Logger.i { "BLE OTA: Service discovered and ready" }
+ }
+ } catch (e: TimeoutCancellationException) {
+ otaService = null
+ currentCoroutineContext().ensureActive()
+ Logger.w { "BLE OTA: Timed out waiting for OTA service discovery. Error: ${e.message}" }
+ throw OtaProtocolException.Timeout("Timed out waiting for BLE OTA service discovery")
+ } catch (e: OtaProtocolException) {
+ otaService = null
+ throw e
+ } catch (e: kotlinx.coroutines.CancellationException) {
+ otaService = null
+ throw e
+ } catch (@Suppress("TooGenericExceptionCaught") e: Exception) {
+ otaService = null
+ Logger.w(e) { "BLE OTA: Failed to prepare OTA service" }
+ throw OtaProtocolException.ConnectionFailed("Failed to prepare BLE OTA service", e)
}
}
@@ -264,6 +323,12 @@ class BleOtaTransport(
}
override suspend fun close() {
+ notificationJob?.cancel()
+ notificationJob = null
+ connectionStateJob?.cancel()
+ connectionStateJob = null
+ otaService = null
+ responseChannel.close()
bleConnection.disconnect()
isConnected = false
transportScope.cancel()
@@ -275,6 +340,7 @@ class BleOtaTransport(
}
private suspend fun writeData(data: ByteArray, writeType: BleWriteType): Int {
+ val service = otaService ?: throw OtaProtocolException.TransferFailed("BLE OTA service is not ready")
// takeIf { it > 0 }: a non-positive negotiated length would stall the loop (offset never advances); fall
// back to a single whole-buffer write instead of looping forever.
val maxLen = bleConnection.maximumWriteValueLength(writeType)?.takeIf { it > 0 } ?: data.size
@@ -286,7 +352,7 @@ class BleOtaTransport(
val chunkSize = minOf(data.size - offset, maxLen)
val packet = data.copyOfRange(offset, offset + chunkSize)
- bleConnection.profile(OTA_SERVICE_UUID) { service -> service.write(otaChar, packet, writeType) }
+ service.write(otaChar, packet, writeType)
offset += chunkSize
packetsSent++
@@ -301,6 +367,23 @@ class BleOtaTransport(
OtaProtocolException.Timeout("Timeout waiting for response after $timeout")
}
+ private fun BleService.requireOtaCharacteristics() {
+ val missing = mutableListOf<String>()
+ if (!hasCharacteristic(txChar)) {
+ missing.add("TX notify characteristic $txChar")
+ }
+ if (!hasCharacteristic(otaChar)) {
+ missing.add("OTA write characteristic $otaChar")
+ }
+
+ if (missing.isNotEmpty()) {
+ throw OtaProtocolException.ConnectionFailed(
+ "ESP32 OTA service was missing required characteristics after BLE service discovery: " +
+ missing.joinToString(separator = "; "),
+ )
+ }
+ }
+
companion object {
private val CONNECTION_TIMEOUT = 15.seconds
private val SUBSCRIPTION_SETTLE = 500.milliseconds
diff --git a/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleScanSupport.kt b/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleScanSupport.kt
index 6fe8fd8b94..6d3608ad61 100644
--- a/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleScanSupport.kt
+++ b/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleScanSupport.kt
@@ -20,7 +20,9 @@ import co.touchlab.kermit.Logger
import kotlinx.coroutines.TimeoutCancellationException
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.delay
+import kotlinx.coroutines.ensureActive
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.firstOrNull
import kotlinx.coroutines.flow.onEach
@@ -79,12 +81,12 @@ internal suspend fun scanForBleDevice(
.scan(timeout = scanTimeout, serviceUuid = serviceUuid)
.onEach { d ->
if (foundDevices.add(d.address)) {
- Logger.d { "$tag: Scan found device: ${d.address} (name=${d.name})" }
+ Logger.d { "$tag: Scan found candidate device (name=${d.name})" }
}
}
.firstOrNull(predicate)
if (device != null) {
- Logger.i { "$tag: Found target device at ${device.address}" }
+ Logger.i { "$tag: Found target device" }
return device
}
Logger.w { "$tag: Target not in ${foundDevices.size} devices found" }
@@ -104,6 +106,7 @@ internal suspend fun scanForBleDevice(
internal suspend fun <T> Channel<T>.receiveWithin(timeout: Duration, onTimeout: () -> Throwable): T = try {
withTimeout(timeout) { receive() }
} catch (@Suppress("SwallowedException") e: TimeoutCancellationException) {
+ currentCoroutineContext().ensureActive()
throw onTimeout()
}
diff --git a/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransportTest.kt b/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransportTest.kt
index 73e487402d..beca5d8d49 100644
--- a/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransportTest.kt
+++ b/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransportTest.kt
@@ -22,6 +22,7 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.test.runTest
import org.meshtastic.core.ble.BleWriteType
import org.meshtastic.core.ble.MeshtasticBleConstants.OTA_NOTIFY_CHARACTERISTIC
+import org.meshtastic.core.ble.MeshtasticBleConstants.OTA_WRITE_CHARACTERISTIC
import org.meshtastic.core.testing.FakeBleConnection
import org.meshtastic.core.testing.FakeBleConnectionFactory
import org.meshtastic.core.testing.FakeBleDevice
@@ -39,7 +40,15 @@ class BleOtaTransportTest {
private fun createTransport(
scanner: FakeBleScanner = FakeBleScanner(),
connection: FakeBleConnection = FakeBleConnection(),
+ seedOtaCharacteristics: Boolean = true,
): Triple<BleOtaTransport, FakeBleScanner, FakeBleConnection> {
+ if (seedOtaCharacteristics) {
+ // Seed at the choke point instead of every connect() site — the new
+ // service.requireOtaCharacteristics() validation in BleOtaTransport.connect() rejects
+ // services missing these, so default to present; negative tests opt out.
+ connection.service.addCharacteristic(OTA_NOTIFY_CHARACTERISTIC)
+ connection.service.addCharacteristic(OTA_WRITE_CHARACTERISTIC)
+ }
val transport =
BleOtaTransport(
scanner = scanner,
@@ -118,6 +127,42 @@ class BleOtaTransportTest {
assertIs<OtaProtocolException.ConnectionFailed>(result.exceptionOrNull())
}
+ @Test
+ fun `connect fails when OTA characteristics are missing`() = runTest {
+ val (transport, scanner) = createTransport(seedOtaCharacteristics = false)
+
+ scanner.emitDevice(FakeBleDevice(address))
+
+ val result = transport.connect()
+
+ assertTrue(result.isFailure)
+ val exception = result.exceptionOrNull()
+ assertIs<OtaProtocolException.ConnectionFailed>(exception)
+ val message = exception.message.orEmpty()
+ assertTrue(message.contains("OTA service"))
+ assertTrue(message.contains("TX notify characteristic"))
+ assertTrue(message.contains("OTA write characteristic"))
+ }
+
+ @Test
+ fun `connect fails when notification observation fails before subscription`() = runTest {
+ val scanner = FakeBleScanner()
+ val connection = FakeBleConnection()
+ val failure = IllegalStateException("observe failed before CCCD")
+ connection.service.observeBeforeSubscriptionExceptionByCharacteristic[OTA_NOTIFY_CHARACTERISTIC] = failure
+ val (transport) = createTransport(scanner, connection)
+
+ scanner.emitDevice(FakeBleDevice(address))
+
+ val result = transport.connect()
+
+ assertTrue(result.isFailure)
+ val exception = result.exceptionOrNull()
+ assertIs<OtaProtocolException.ConnectionFailed>(exception)
+ val cause = assertIs<IllegalStateException>(exception.cause)
+ assertEquals(failure.message, cause.message)
+ }
+
// -----------------------------------------------------------------------
// startOta()
// -----------------------------------------------------------------------
@@ -249,6 +294,34 @@ class BleOtaTransportTest {
assertEquals(1.0f, progressValues.last())
}
+ @Test
+ fun `streamFirmware reuses discovered OTA service for chunk writes`() = runTest {
+ val scanner = FakeBleScanner()
+ val connection = FakeBleConnection()
+ val (transport) = createTransport(scanner, connection)
+ connection.maxWriteValueLength = 200
+ scanner.emitDevice(FakeBleDevice(address))
+ transport.connect().getOrThrow()
+ val profileCallsAfterConnect = connection.profileCalls
+
+ emitResponse(connection, "OK")
+ transport.startOta(600L, "hash").getOrThrow()
+
+ emitResponse(connection, "ACK")
+ emitResponse(connection, "ACK")
+ emitResponse(connection, "OK")
+
+ val result = transport.streamFirmware(ByteArray(600) { it.toByte() }, 512) {}
+
+ assertTrue(result.isSuccess, "streamFirmware failed: ${result.exceptionOrNull()}")
+ assertEquals(1, profileCallsAfterConnect, "connect should discover the OTA profile once")
+ assertEquals(
+ profileCallsAfterConnect,
+ connection.profileCalls,
+ "writes should reuse the discovered OTA service",
+ )
+ }
+
@Test
fun `streamFirmware fails on connection lost`() = runTest {
val scanner = FakeBleScanner()
Served by rngit 1.5.2 - Generated in 0.09s